package com.shujia.base;

import com.shujia.utils.BaseUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CellScanner;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentSkipListMap;

public class Demo1 {
    private Connection conn;
    private Admin admin;

    @Before
    public void getConnection() {
        try {
            Configuration conf = HBaseConfiguration.create();
            conf.set("hbase.zookeeper.quorum", "hadoop102:2181,hadoop103:2181,hadoop104:2181");
            conn = ConnectionFactory.createConnection(conf);
            admin = conn.getAdmin();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void createTable() {
        BaseUtils.createTable(admin,"students_1",Bytes.toBytes("info"));
    }

    @Test
    public void insertDataToTable() {
        BufferedReader br = null;
        Table table = null;
        try {
            TableName tn = TableName.valueOf("students_1");
            table = conn.getTable(tn);
            ArrayList<Put> puts = new ArrayList<>();
            br = new BufferedReader(new FileReader("src/main/resources/students.csv"));
            String line;
            while ((line = br.readLine()) != null) {
                String[] split = line.split(",");
                puts.add(new Put(Bytes.toBytes(split[0])).addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes(split[1])));
                puts.add(new Put(Bytes.toBytes(split[0])).addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes(split[2])));
                puts.add(new Put(Bytes.toBytes(split[0])).addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes(split[3])));
                puts.add(new Put(Bytes.toBytes(split[0])).addColumn(Bytes.toBytes("info"), Bytes.toBytes("clazz"), Bytes.toBytes(split[4])));
            }
            table.put(puts);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (br != null) br.close();
                if (table != null) table.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    @Test
    public void alterTable() {
        TableName students = TableName.valueOf("students");
        ColumnFamilyDescriptor score = ColumnFamilyDescriptorBuilder.newBuilder("score".getBytes()).build();
        try {
            admin.addColumnFamily(students, score);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Test
    public void showAllTable() {
        try {
            TableName[] tableNames = admin.listTableNames();
            System.out.println(Arrays.toString(tableNames));

        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }

    @Test
    public void showTableData() {
        BaseUtils.printAll("students_1",conn,admin);
//        TableName students1 = TableName.valueOf("students_1");
//        try {
//            TableDescriptor descriptor = admin.getDescriptor(students1);
//            Table table = conn.getTable(students1);
//            ResultScanner scanner = table.getScanner(new Scan());
//            scanner.forEach(row -> {
//                StringBuffer res = new StringBuffer();
//                String rowkey =new String(row.getRow());
//                for (byte[] columnFamilyName : descriptor.getColumnFamilyNames()) {
//                    NavigableMap<byte[], byte[]> familyMap = row.getFamilyMap(columnFamilyName);
//                    res.append(rowkey).append("\t").append(new String(columnFamilyName)).append(":{");
//                    familyMap.forEach((k, v) -> {
//                        res.append(new String(k)).append("=").append(new String(v)).append(",");
//                    });
//                    res.delete(res.length() - 1, res.length());
//                    res.append("};");
//                    System.out.println(res);
//                }
//            });
//        } catch (IOException e) {
//            throw new RuntimeException(e);
//        }
    }

    /*
        查询文科一班学生总分排名前10的学生（输出：学号，姓名，班级，总分）结果写到hbase
     */
    @Test
    public void scan_homework() {

    }
    @After
    public void close() {
        try {
            if (conn != null) conn.close();
            if (admin != null) admin.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
